package com.imooc.spark.streaming.windowOperations

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 当在（K，V）对的DStream上调用时，返回（K，V）对的新DStream，
  * 其中使用给定的reduce函数func 在滑动窗口中的批次聚合每个键的值。
  * 注意：默认情况下，这使用Spark的默认并行任务数（本地模式为2，在群集模式下，数量由config属性确定spark.default.parallelism）进行分组。
  * 您可以传递可选 numTasks参数来设置不同数量的任务。
  *
  * @description reduceByKeyAndWindow（func，windowLength，slideInterval，[ numTasks ]）
  * @author yuyon26@126.com
  * @date 2018/10/24 20:12
  */
object ReduceByKeyAndWindow {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint("D:\\spark\\checkpoint")
    ssc.sparkContext.setLogLevel("ERROR")

    val lines = ssc.socketTextStream("192.168.31.30", 9999)

    val fmapLines = lines.flatMap(_.split(" "))
    val tuple = fmapLines.map(word => (word, 1))
    val reduceByKeyAndWindow = tuple.reduceByKeyAndWindow((a:Int,b:Int) => (a + b),Seconds(10),Seconds(5))

    reduceByKeyAndWindow.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
